package com.foreveross.fetl.queue;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Calendar;
import java.util.Collection;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.UUID;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.log4j.Logger;

import com.foreveross.fetl.common.HBaseTable;
import com.foreveross.fetl.common.hbase.client.HBaseHelper;
import com.foreveross.fetl.core.domain.task.SourceEntity;
import com.foreveross.fetl.queue.util.LogMessageUtils;
import com.foreveross.fetl.util.DatabaseBeanUtils;
import com.foreveross.fetl.util.DateUtils;
import com.foreveross.fetl.util.RowkeyUtils;

public class GenerateFlightInfo {

	private static Logger log = Logger.getLogger(GenerateFlightInfo.class);
	
	private int tasksNumber;//任务数 多少条rowkey
	
	private int failuresNumber;//失败数
	
	
	private int insertNumber; //插入条数 
	
	private long excuteStartTime;
	
	private String taskTime=null;
	
	public void  rollBackNumber(){
		tasksNumber=0;
		failuresNumber=0;
		insertNumber=0;
		excuteStartTime = System.currentTimeMillis();
		LogMessageUtils.create("生成航班信息", this.getClass().getSimpleName());
		
	}
	
	/**
	 * @param args
	 */
	public static void main(String[] args) {
		GenerateFlightInfo o = new GenerateFlightInfo();
		Calendar start = Calendar.getInstance();
		start.set(2013, 02, 27, 16, 00);
		Calendar end = Calendar.getInstance();
		end.set(2013, 02, 28, 14, 00);
		o.process(start.getTime(), end.getTime());
	}

	/**
	 * 定時線程的調用入口
	 */
	public void execute() {
		rollBackNumber();
		long start = System.currentTimeMillis();
		log.info("generateFlightInfo start:"
				+ DateUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss.SSS"));
		Calendar c = Calendar.getInstance();
		c.add(Calendar.HOUR_OF_DAY, -1);
		taskTime=DateUtils.format(c.getTime(), "yyyy-MM-dd HH");
		try {
			generateFlightInfo(c.getTime(), new Date());
		} catch (Exception e) {
			e.printStackTrace();
			LogMessageUtils.build("生成航班信息", this.getClass().getSimpleName(), 3, 
					com.foreveross.fetl.util.StringUtils.getStackTrace(e), failuresNumber, tasksNumber, 
					insertNumber, taskTime, excuteStartTime);
		}
		LogMessageUtils.build("生成航班信息", this.getClass().getSimpleName(), 2, null, 
				failuresNumber, tasksNumber, insertNumber, taskTime, excuteStartTime);
		
		log.info("generateFlightInfo end："
				+ DateUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss.SSS"));
		log.info("total spend time:" + (System.currentTimeMillis() - start));
	}
	
	public void manualExecute(final String start,String time){
		this.rollBackNumber();
		try {
		List<String> timeList = Arrays.asList(time.split(";"));
		Calendar c = Calendar.getInstance();
		taskTime = start + " " +time;
		for (String s : timeList) {
			String str =start;
			str += s;
			Date startTime = DateUtils.parse(str, "yyyyMMddHH");
			c.setTime(startTime);
			c.add(Calendar.HOUR_OF_DAY, +1);
			Date endTime = c.getTime();
			this.process(startTime, endTime);
			}
		} catch (Exception e) {
			LogMessageUtils.build("生成航班信息", this.getClass().getSimpleName(), 3, com.foreveross.fetl.util.StringUtils.getStackTrace(e), failuresNumber, tasksNumber, 
					insertNumber, taskTime, excuteStartTime);
		}
		LogMessageUtils.build("生成航班信息", this.getClass().getSimpleName(), 2, null, 
				failuresNumber, tasksNumber, insertNumber, taskTime, excuteStartTime);
	}

	/**
	 * 将抓取来的航班信息保存到数据库中
	 * 
	 * @param endTime
	 * @param startTime
	 */
	public void process(Date startTime, Date endTime) {
		generateFlightInfo(startTime, endTime);
	}

	public static final String querySQL = "select ID from fdc_flightinfo i where i.COMPANY_CODE=? and i.flightno=? and i.FLIGHT_START_TIME=? and i.FROMCITY=? and i.TOCITY=? ";//
	public static final String insertSQL = "insert into fdc_flightinfo(ID,FROMCITY,FROMCITYNAME,TOCITY,TOCITYNAME,FLIGHTNO, AREA_CODE,AREA_NAME,COMPANY_CODE,  COMPANY_NAME,FLIGHT_START_TIME, FLIGHT_END_TIME,ENTRY_TIME) values(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,?)";
	public static final String updateSQL = "update fdc_flightinfo set FLIGHT_START_TIME = ?,  FLIGHT_END_TIME = ?,ENTRY_TIME = ? where ID=?";
	/**
	 * 将抓取来的航班信息保存到数据库中
	 * 
	 * @param endTime
	 * @param startTime
	 */
	public void generateFlightInfo(Date startTime, Date endTime) {
	
		String startStr = DateUtils.format(startTime, "yyyyMMddHH");
		Connection conn = null;
		ResultScanner scan = null;
		SourceEntity source = new SourceEntity();
		List<SourceEntity> listHome=source.findSourceEntityByType(SourceEntity.TYPE_HOME);
		List<String> strchannels= new ArrayList<String>();
		if(null!=listHome){
			for(SourceEntity entity:listHome){
				strchannels.add(entity.getHbaseKey());
			}
		}
		Collection<String> channels = strchannels;
		for (final String channel : channels) {
			
				StringBuffer startRowKey = new StringBuffer(RowkeyUtils.getRowKey(startStr, channel, RowkeyUtils.START_ROW_KEY ));
				
				StringBuffer endRowKey1 = new StringBuffer(RowkeyUtils.getRowKey(startStr, channel, RowkeyUtils.STOP_ROW_KEY ));
				try {
					conn = DatabaseBeanUtils.getConnection();
					conn.setAutoCommit(false);
					log.info("generateFlightInfo startRowKey:" + startRowKey.toString()
							+ "  endRowkey:" + endRowKey1.toString());
					scan = HBaseHelper.getRecords(
							HBaseTable.BASE_DATA_FLIGHT,
							startRowKey.toString(),
							endRowKey1.toString(),new String[]{"planeinfo","transitsInfo"},null);
					long l = 0;
					String rowid =null;
					for (Result r : scan) {
						try {
							rowid=new String(r.getRow());
							tasksNumber++;
						
							Map<String, Map<String, String>> map = HBaseHelper
									.transformHBaseRowToMap(r);
							Map<String ,String> transitsMap=map.get("transitsInfo");
							if(null !=transitsMap && transitsMap.containsKey("flightNo_1")){
								continue;
							}
						
							Map<String, String> planeinfoMap = map.get("planeinfo");
							log.info("flight insert into:"+new String(r.getRow(),"utf-8")+" flightNo:"+planeinfoMap.get("flightId"));
							if(isNumeric(planeinfoMap.get("flightId"))){
								continue;
							}
							if(StringUtils.isNotBlank(planeinfoMap.get("carrierKey"))
									&& StringUtils.isNotBlank(planeinfoMap.get("flightId"))){
								this.genOneFlight(planeinfoMap, conn,rowid);
								l++;
							}
							if (l % 100 == 0) {
							//	System.out.println(l);
								log.info("commit datas:"+l);
								conn.commit();
							}
						}catch(Exception e){
							failuresNumber++;
							log.error("insert flight error "+e.getMessage());
						}
					}
					
					try{
						conn.commit();
					}catch(Exception e){
						failuresNumber++;
						log.error("insert flight error "+e.getMessage());
					}
				}catch (Exception e) {
				
					e.printStackTrace();
				} finally {
					if(scan != null) scan.close();
					scan = null;
					closeResource(null, null, conn);
				}
			
		}
	}

	private void genOneFlight(Map<String, String> planeinfoMap, Connection conn, String rowid){
		ResultSet rs = null;
		PreparedStatement stmt = null;
		String id = null;
		try{
			if(null!=planeinfoMap.get("flightId") && planeinfoMap.get("flightId").indexOf("|")>0){
				planeinfoMap.put("flightId", planeinfoMap.get("flightId").substring(0, planeinfoMap.get("flightId").indexOf("|")));
			}
			if(planeinfoMap.get("flightId").length()>32){
				return;
			}
			String flightBatchStartTime = planeinfoMap.get("flightBatchStartTime").substring(8, 10)+ ":"+ planeinfoMap.get("flightBatchStartTime").substring(10, 12);
			String flightBatchEndTime = planeinfoMap.get(
					"flightBatchEndTime").substring(8, 10)+ ":"+ planeinfoMap.get("flightBatchEndTime").substring(10, 12);
			int i = 1;
		
			try{
				stmt = conn.prepareStatement(querySQL);
				stmt.setString(i++, planeinfoMap.get("carrierKey"));
				stmt.setString(i++, planeinfoMap.get("flightId"));
				stmt.setString(i++, flightBatchStartTime);
				stmt.setString(i++, planeinfoMap.get("fromCity"));
				stmt.setString(i++, planeinfoMap.get("toCity"));
				rs = stmt.executeQuery();
				if(rs.next()){
					id = rs.getString("ID");
				}
			}catch(Exception e){
				failuresNumber++;
				log.error(rowid);
				log.error("insert flight error "+com.foreveross.fetl.util.StringUtils.getStackTrace(e));
			}finally{
				closeResource(rs, stmt);
			}

			i = 1;
			if(StringUtils.isNotBlank(id)){//更新
				try{
					stmt = conn.prepareStatement(updateSQL);
					stmt.setString(i++, flightBatchStartTime);
					stmt.setString(i++, flightBatchEndTime);
					stmt.setTimestamp(i++,
							new Timestamp(System.currentTimeMillis()));
					stmt.setString(i++, id);
					stmt.executeUpdate();
					insertNumber++;
				}catch(Exception e){
					failuresNumber++;
					log.error(rowid);
					log.error("insert flight error "+com.foreveross.fetl.util.StringUtils.getStackTrace(e));
				}finally{
					closeResource(rs, stmt);
				}
			}else{//插入
				try{
					id = UUID.randomUUID().toString().replaceAll("-", "");
					stmt = conn.prepareStatement(insertSQL);
					stmt.setString(i++, id);
					
					stmt.setString(i++, planeinfoMap.get("fromCity"));
					stmt.setString(i++, planeinfoMap.get("fromCityName"));
					stmt.setString(i++, planeinfoMap.get("toCity"));
					stmt.setString(i++, planeinfoMap.get("toCityName"));
					stmt.setString(i++, planeinfoMap.get("flightId"));
					stmt.setString(i++, planeinfoMap.get("areaCode"));
					stmt.setString(i++, planeinfoMap.get("areaName"));
					stmt.setString(i++, planeinfoMap.get("carrierKey"));
					stmt.setString(i++, planeinfoMap.get("carrierFullName"));
					stmt.setString(i++, flightBatchStartTime);
					stmt.setString(i++, flightBatchEndTime);
					stmt.setTimestamp(i++,
							new Timestamp(System.currentTimeMillis()));
					stmt.execute();
					insertNumber++;
				}catch(Exception e){
					failuresNumber++;
					log.error(rowid);
					log.error("insert flight error "+com.foreveross.fetl.util.StringUtils.getStackTrace(e));
				}finally{
					closeResource(rs, stmt);
				}
			}
//			if(rowid.startsWith("20020")){
//				log.error(rowid);
//			}
		}catch(Exception e){
			failuresNumber++;
			log.error(rowid);
			log.error("insert flight error "+com.foreveross.fetl.util.StringUtils.getStackTrace(e));
		}finally{
			closeResource(rs, stmt);
		}
	}
	/**
	 * 关闭连接资源
	 * 
	 * @param rs
	 * @param stmt
	 * @param conn
	 */
	public void closeResource(ResultSet rs, PreparedStatement stmt,
			Connection conn) {
			if (rs != null) {
				try {
					rs.close();
				} catch (SQLException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
				rs = null;
			}
			if (stmt != null) {
				try {
					stmt.close();
				} catch (SQLException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
				stmt = null;
			}
			if (conn != null) {
				try {
					conn.close();
				} catch (SQLException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
				conn = null;
			}
	}
	public void closeResource(ResultSet rs, PreparedStatement stmt) {
		
			if (rs != null) {
				try {
					rs.close();
				} catch (SQLException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
				rs = null;
			}
			if (stmt != null) {
				try {
					stmt.close();
				} catch (SQLException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
				stmt = null;
			}
	}
	

	
	/**
	 * 判断是否为数字    是 true
	 * @param str
	 * @return
	 */
	private   boolean isNumeric(String str){
			return str.matches("\\d*");    
		 } 

}
